package com.chenjj.bigdata.flink.window.trigger

import org.apache.flink.api.common.functions.ReduceFunction
import org.apache.flink.api.common.state.ReducingStateDescriptor
import org.apache.flink.api.scala.typeutils.Types
import org.apache.flink.streaming.api.windowing.triggers.{Trigger, TriggerResult}
import org.apache.flink.streaming.api.windowing.windows.TimeWindow

/**
  * 实现窗口触发的时间间隔，当窗口是Session Window时，如果用户长时间不停的操作，导致session gap一直都不生成
  * 因此该用户的数据会长期存储在窗口中，如果需要至少每隔5分钟统计一下窗口的结果，则需要自定义Trigger。
  *
  * EarlyTriggeringTrigger就是用来实现这个需求
  */
class ContinuousEventTimeTrigger(interval:Long) extends Trigger[Object,TimeWindow]{
  /**
    * 重定义Java.lang.Long类型未JLong类型
    */
  private type JLong = java.lang.Long

  /**
    * 实现函数，求取2个时间戳的最小值
    */
  private val min = new ReduceFunction[JLong] {
    override def reduce(value1: JLong, value2: JLong): JLong = Math.min(value1,value2)
  }

  private val stateDesc = new ReducingStateDescriptor[JLong]("trigger-time",min,Types.LONG)


  override def onElement(element: Object, timestamp: Long, window: TimeWindow, ctx: Trigger.TriggerContext): TriggerResult = ???

  override def onProcessingTime(time: Long, window: TimeWindow, ctx: Trigger.TriggerContext): TriggerResult = ???

  override def onEventTime(time: Long, window: TimeWindow, ctx: Trigger.TriggerContext): TriggerResult = ???

  override def clear(window: TimeWindow, ctx: Trigger.TriggerContext): Unit = ???
}
